package com.tdy.cdc.app;

import com.tdy.cdc.app.dwd.App10_DwdMhsCarePlan;
import com.tdy.cdc.util.ConfigUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author NanHuang
 * @Date 2023/1/24
 */
public abstract class BaseSqlApp {
    public void init(Integer port,String groupId,Integer... params){
        System.setProperty("HADOOP_USER_NAME","root");
        // 1 创建编程环境
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",port);//设置前web端口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        if (params.length == 1) {
            env.setParallelism(params[0]);//设置并行度
        }
        // 2 设置状态后端和Checkpoint
        env.setStateBackend(new EmbeddedRocksDBStateBackend());//启用RocksDB状态后端
        env.enableCheckpointing(60 * 1000);//设置checkpoint的周期
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);//设置同时checkpoint的数量
        env.getCheckpointConfig().setCheckpointTimeout(10 * 60 * 1000);//checkpoint的超时时间
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setCheckpointStorage(ConfigUtil.getProperty("checkpoint.storage.url.prefix.test") + groupId);//设置checkpoint的存储路径
        // 3 对数据进行处理
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.getConfig().getConfiguration().setString("pipeline.name",groupId);
        invoke(env,tableEnv);
    }

    protected abstract void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv);

    public DataStreamSource<String> readKafka(StreamExecutionEnvironment env, String topicName,String groupId){
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers(ConfigUtil.getProperty("kafka.bootstrap.servers"))
                .setTopics(topicName)
                .setGroupId(groupId)
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();
        return env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
    }

    public void readTdyListDiag(StreamTableEnvironment tableEnv, String groupId){
        tableEnv.executeSql("create table tdy_list_diag (" +
                "pk                     string," +
                "unique_id              string," +
                "domain_id              string," +
                "xds_type               string," +
                "xds_version            string," +
                "diag_code              string," +
                "diag_name              string," +
                "diag_date              string," +
                "diag_institution       string," +
                "diag_order             string," +
                "diag_outcome_code      string," +
                "diag_outcome_name      string," +
                "diag_scode             string," +
                "diag_sname             string," +
                "admission_code         string," +
                "admission_name         string," +
                "diag_type              string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("tdy_list_diag",groupId));
    }

    public void readTdyListOper(StreamTableEnvironment tableEnv, String groupId){
        tableEnv.executeSql("create table tdy_list_oper (" +
                "pk                     string," +
                "unique_id              string," +
                "domain_id              string," +
                "xds_type               string," +
                "xds_version            string," +
                "oper_code              string," +
                "oper_name              string," +
                "oper_date              string," +
                "oper_cut_code          string," +
                "oper_cut_name          string," +
                "oper_ihg_code          string," +
                "oper_ihg_name          string," +
                "oper_level             string," +
                "oper_level_code        string," +
                "oper_method            string," +
                "oper_num               string," +
                "oper_post              string," +
                "oper_post_code         string," +
                "oper_room              string," +
                "oper_start_date        string," +
                "oper_end_date          string," +
                "admit_room_date        string," +
                "out_oper_date          string," +
                "preparations           string," +
                "surgeon_code           string," +
                "surgeon_name           string," +
                "target_code            string," +
                "target_name            string," +
                "anes_method_code       string," +
                "anes_method_name       string," +
                "anes_obs               string," +
                "anesth_code            string," +
                "anesth_name            string," +
                "anesth_western         string," +
                "anesth_western_sign    string," +
                "assistant_i            string," +
                "assistant_icode        string," +
                "assistant_ii           string," +
                "assistant_iicode       string," +
                "in_pathology_mark      string," +
                "intervention           string," +
                "treatment_process      string," +
                "pro_process            string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("tdy_list_oper",groupId));
    }

    public void readTdyListOrder(StreamTableEnvironment tableEnv, String groupId){
        tableEnv.executeSql("create table tdy_list_order (" +
                "pk                     string," +
                "unique_id              string," +
                "domain_id              string," +
                "xds_type               string," +
                "xds_version            string," +
                "audit_code             string," +
                "audit_date             string," +
                "audit_order            string," +
                "cancel_code            string," +
                "cancel_date            string," +
                "cancel_name            string," +
                "check_name             string," +
                "check_code             string," +
                "check_date             string," +
                "end_date               string," +
                "executor_code          string," +
                "executor_date          string," +
                "executor_dept          string," +
                "executor_name          string," +
                "executor_status        string," +
                "open_code              string," +
                "open_date              string," +
                "open_dept              string," +
                "open_name              string," +
                "order_code             string," +
                "order_content          string," +
                "order_remark           string," +
                "order_type             string," +
                "pdrug_group            string," +
                "start_date             string," +
                "stop_code              string," +
                "stop_date              string," +
                "stop_name              string," +
                "appy_id                string," +
                "order_category_code    string," +
                "order_category         string," +
                "drug_dose              string," +
                "dose_unit              string," +
                "drug_usage             string," +
                "drug_frequency         string," +
                "recipel_days           string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("tdy_list_order",groupId));
    }

    public void readTdyListLexam(StreamTableEnvironment tableEnv, String groupId){
        tableEnv.executeSql("create table tdy_list_lexam (" +
                "pk              string," +
                "unique_id       string," +
                "domain_id       string," +
                "xds_type        string," +
                "xds_version     string," +
                "item_code       string," +
                "item_name       string," +
                "result_name     string," +
                "result_code     string," +
                "quanty_result   string," +
                "quanty_unit     string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("tdy_list_lexam",groupId));
    }

    public void readTdyListExam(StreamTableEnvironment tableEnv, String groupId){
        tableEnv.executeSql("create table tdy_list_exam (" +
                "pk                        string," +
                "unique_id                 string," +
                "domain_id                 string," +
                "xds_type                  string," +
                "xds_version               string," +
                "check_date                string," +
                "item_name                 string," +
                "item_code                 string," +
                "quanty_result             string," +
                "quanty_unit               string," +
                "result_code               string," +
                "result_name               string," +
                "sample_date               string," +
                "receipt_date              string," +
                "specimen_category         string," +
                "specimen_fixative         string," +
                "specimen_status           string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("tdy_list_exam",groupId));
    }

    public void readTdyListTest(StreamTableEnvironment tableEnv, String groupId){
        tableEnv.executeSql("create table tdy_list_test (" +
                "pk                       string," +
                "unique_id                string," +
                "domain_id                string," +
                "xds_type                 string," +
                "xds_version              string," +
                "inspect_date             string," +
                "item_name                string," +
                "item_code                string," +
                "quanty_result            string," +
                "quanty_unit              string," +
                "receipt_date             string," +
                "result_code              string," +
                "result_name              string," +
                "sample_date              string," +
                "specimen_category        string," +
                "specimen_status          string," +
                "reference_value          string," +
                "result_sign              string," +
                "quality_name             string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("tdy_list_test",groupId));
    }

    protected void readTdyListNurseTrachea(StreamTableEnvironment tableEnv, String groupId) {
        tableEnv.executeSql("create table tdy_list_nurse_trachea (" +
                "pk           String,"+
                "unique_id    String,"+
                "domain_id    String,"+
                "xds_type     String,"+
                "xds_version  String,"+
                "code         String,"+
                "name         String"+
                ")"+
                FlinkSourceUtil.getKafkaWith("tdy_list_nurse_trachea",groupId));
    }

    protected void readTdyListNurseSafety(StreamTableEnvironment tableEnv, String groupId) {
        tableEnv.executeSql("create table tdy_list_nurse_safety (" +
                "pk          String,"+
                "unique_id   String,"+
                "domain_id   String,"+
                "xds_type    String,"+
                "xds_version String,"+
                "name        String,"+
                "code        String"+
                ")"+
                FlinkSourceUtil.getKafkaWith("tdy_list_nurse_safety",groupId));
    }

    protected void readTdyListDietGuide(StreamTableEnvironment tableEnv, String groupId) {
        tableEnv.executeSql("create table tdy_list_diet_guide (" +
                "pk          String,"+
                "unique_id   String,"+
                "domain_id   String,"+
                "xds_type    String,"+
                "xds_version String,"+
                "code        String,"+
                "name        String"+
                ")"+
                FlinkSourceUtil.getKafkaWith("tdy_list_diet_guide",groupId));
    }

    protected void readTdyListHandover(StreamTableEnvironment tableEnv, String groupId) {
        tableEnv.executeSql("create table tdy_list_handover (" +
                "pk                    String,"+
                "unique_id             String,"+
                "domain_id             String,"+
                "xds_type              String,"+
                "xds_version           String,"+
                "handover_checkup      String,"+
                "handover_date         String,"+
                "handover_nurse        String,"+
                "handover_nursecode    String,"+
                "shipper_code          String,"+
                "shipper_name          String "+
                ")" +
                FlinkSourceUtil.getKafkaWith("tdy_list_handover","App09_DwdMhsSurgicalNursingRecord"));
    }

    protected void readPatient(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table patient (" +
                "`PK`                                    bigint, " +
                "`GLOBAL_ID`                             string, " +
                "`DOMAIN_ID`                             string, " +
                "`MESSAGE_CREATE_TIME`                   string, " +
                "`PATIENT_ID`                            string, " +
                "`PATIENT_NAME`                          string, " +
                "`NAME_SPELL`                            string, " +
                "`DATE_OF_BIRTH`                         string, " +
                "`IDENTITY_NO`                           string, " +
                "`HOME_PLACE`                            string, " +
                "`REGISTERED_PLACE`                      string, " +
                "`NATIVE_PLACE`                          string, " +
                "`BIRTH_PLACE`                           string, " +
                "`SOCIAL_SECURITY_NUMBER`                string, " +
                "`HEALTH_CARD_NUMBER`                    string, " +
                "`CITIZEN_CARD_NUMBER`                   string, " +
                "`SEX_CODE`                              string, " +
                "`SEX`                                   string, " +
                "`ETHNIC_CODE`                           string, " +
                "`ETHNIC`                                string, " +
                "`COUNTRY_CODE`                          string, " +
                "`COUNTRY`                               string, " +
                "`MEDICAL_INSURANCE_NUMBER`              string, " +
                "`ELDER_CERTIFICATE_NUMBER`              string, " +
                "`MEDICAL_RECORDS_NUMBER`                string, " +
                "`LANGUAGE_CODE`                         string, " +
                "`LANGUAGE`                              string, " +
                "`RELIGION_CODE`                         string, " +
                "`RELIGION`                              string, " +
                "`MARITAL_CODE`                          string, " +
                "`MARITAL`                               string, " +
                "`DEGREE_CODE`                           string, " +
                "`DEGREE`                                string, " +
                "`HOME_PHONE`                            string, " +
                "`HOME_ZIP_CODE`                         string, " +
                "`REGISTERED_ZIP_CODE`                   string, " +
                "`BIRTH_ZIP_CODE`                        string, " +
                "`NATIVE_ZIP_CODE`                       string, " +
                "`EMAIL`                                 string, " +
                "`IDENTITY`                              string, " +
                "`IDENTITY_CODE`                         string, " +
                "`WORK_ADDRESS`                          string, " +
                "`WORK_PHONE`                            string, " +
                "`CONTACT_RELATIONSHIP_CODE`             string, " +
                "`CONTACT_RELATIONSHIP`                  string, " +
                "`RELATION_ADDRESS`                      string, " +
                "`RELATION_NAME`                         string, " +
                "`RELATION_PHONE`                        string, " +
                "`RELATION_ZIP_CODE`                     string, " +
                "`BLOOD_TYPE_CODE`                       string, " +
                "`BLOOD_TYPE`                            string, " +
                "`CERTIFICATES_TYPE_CODE`                string, " +
                "`CERTIFICATES_NUMBER`                   string, " +
                "`CHARGE_TYPE`                           string, " +
                "`ALLERGY_INFORMATION`                   string, " +
                "`INSURANCE_TYPE`                        string, " +
                "`DEATH_TIME`                            string, " +
                "`DEATH_IND`                             string, " +
                "`INSURANCE_TYPE_CODE`                   string, " +
                "`EFFECTIVE_TIME`                        string, " +
                "`CERTIFICATES_TYPE_DEPICT`              string, " +
                "`PROVINCE`                              string, " +
                "`CITY`                                  string, " +
                "`DISTRICT`                              string, " +
                "`RURAL`                                 string, " +
                "`VILLAGE`                               string, " +
                "`HOUSE_NUMBER`                          string, " +
                "`OCCUPATION_ID`                         string, " +
                "`OCCUPATION_DEPICT`                     string, " +
                "`COMPANY`                               string, " +
                "`HEALTH_CARD_ORGANIZATION_CODE`         string, " +
                "`HEALTH_FILE_NUMBER`                    string, " +
                "`ARCHIVING_ORGANIZATION_CODE`           string, " +
                "`PROVIDER_ORGANIZATION_ID`              string, " +
                "`PROVIDER_ORGANIZATION_NAME`            string, " +
                "`HOME_PROVINCE`                         string, " +
                "`HOME_CITY`                             string, " +
                "`HOME_COUNTY`                           string, " +
                "`HOME_RURAL`                            string, " +
                "`HOME_TOWNSHIP`                         string, " +
                "`HOME_DISTRICT`                         string, " +
                "`HOME_VILLAGE`                          string, " +
                "`HOME_STREET_NAME`                      string, " +
                "`HOME_HOUSE_NUMBER`                     string, " +
                "`REGISTER_PROVINCE`                     string, " +
                "`REGISTER_CITY`                         string, " +
                "`REGISTER_COUNTY`                       string, " +
                "`REGISTER_TOWNSHIP`                     string, " +
                "`REGISTER_STREET_NAME`                  string, " +
                "`REGISTER_DISTRICT`                     string, " +
                "`REGISTER_RURAL`                        string, " +
                "`REGISTER_VILLAGE`                      string, " +
                "`REGISTER_HOUSE_NUMBER`                 string, " +
                "`NATIVE_PROVINCE`                       string, " +
                "`NATIVE_CITY`                           string, " +
                "`WORK_ZIP_CODE`                         string, " +
                "`WORK_PROVINCE`                         string, " +
                "`WORK_CITY`                             string, " +
                "`WORK_COUNTY`                           string, " +
                "`WORK_RURAL`                            string, " +
                "`WORK_TOWNSHIP`                         string, " +
                "`WORK_DISTRICT`                         string, " +
                "`WORK_VILLAGE`                          string, " +
                "`WORK_STREET_NAME`                      string, " +
                "`WORK_HOUSE_NUMBER`                     string, " +
                "`BIRTH_PROVINCE`                        string, " +
                "`BIRTH_CITY`                            string, " +
                "`BIRTH_COUNTY`                          string, " +
                "`BIRTH_RURAL`                           string, " +
                "`BIRTH_TOWNSHIP`                        string, " +
                "`BIRTH_DISTRICT`                        string, " +
                "`BIRTH_VILLAGE`                         string, " +
                "`BIRTH_STREET_NAME`                     string, " +
                "`BIRTH_HOUSE_NUMBER`                    string, " +
                "`RELATION_VILLAGE`                      string, " +
                "`RELATION_PROVINCE`                     string, " +
                "`RELATION_HOUSE_NUMBER`                 string, " +
                "`RELATION_CITY`                         string, " +
                "`RELATION_COUNTY`                       string, " +
                "`RELATION_RURAL`                        string, " +
                "`AGE`                                   string, " +
                "`AGE_UNIT`                              string, " +
                "`PHONE_NUMBER`                          string, " +
                "`STATUS_CODE`                           string, " +
                "`GENETIC_DISEASE_HISTORY`               string, " +
                "`OPERATION_HISTORY`                     string, " +
                "`CREATE_OPERATOR`                       string, " +
                "`HEALTH_CARD_CREATE_TIME`               string, " +
                "`HEALTH_CARD_END_DATE`                  string, " +
                "`COOPERATIVE_NO`                        string, " +
                "`REGISTER_ID`                           string, " +
                "`REGISTER_NAME`                         string, " +
                "`CREATE_TIME`                           timestamp, " +
                "`PUSH_STATUS`                           char, " +
                "`OLD_GLOBAL_ID`                         string, " +
                "`MERGE_STATUS`                          boolean" +
                ")" +
                "with (" +
                "   'connector'='jdbc'," +
                "   'url'='jdbc:mysql://192.168.31.206:3306/test_yuan'," +
                "   'table-name'='patient'," +
                "   'username'='root'," +
                "   'password'='password'," +
                // 设置数据的缓存事件，小时：hour，天：day
                "   'lookup.cache.ttl'='30 s'," +
                // 设置数据的缓存条数
                "   'lookup.cache.max-rows'='100'" +
                ")");
    }

    protected void readVisit(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table visit (" +
                "`PK`                                 bigint, " +
                "`PATIENT_ID`                         string, " +
                "`PATIENT_DOMAIN_ID`                  string, " +
                "`PATIENT_GLOBAL_ID`                  string, " +
                "`VISIT_ID`                           string, " +
                "`OUTPATIENT_ID`                      string, " +
                "`INPATIENT_ID`                       string, " +
                "`OUTPATIENT_VISIT_FLOW_ID`           string, " +
                "`INPATIENT_VISIT_FLOW_ID`            string, " +
                "`INPATIENT_COUNT`                    string, " +
                "`VISIT_DOMAIN_ID`                    string, " +
                "`PATIENT_CATEGORY`                   string, " +
                "`VISIT_AGE`                          string, " +
                "`ADMIT_DEPT_CODE`                    string, " +
                "`ADMIT_DEPT_NAME`                    string, " +
                "`ADMIT_DIAGNOSE_CODE`                string, " +
                "`ADMIT_DIAGNOSE_NAME`                string, " +
                "`ADMIT_DATE`                         string, " +
                "`ADMISSION_DEPT_CODE`                string, " +
                "`ADMISSION_DEPT_NAME`                string, " +
                "`ADMISSION_DIAGNOSE_CODE`            string, " +
                "`ADMISSION_DIAGNOSE_NAME`            string, " +
                "`ADMISSION_DATE`                     string, " +
                "`LEAVE_DEPT_CODE`                    string, " +
                "`LEAVE_DEPT_NAME`                    string, " +
                "`LEAVE_DIAGNOSE_CODE`                string, " +
                "`LEAVE_DIAGNOSE_NAME`                string, " +
                "`LEAVE_DATE`                         string, " +
                "`EMERGENCY_DEPT_CODE`                string, " +
                "`EMERGENCY_DEPT_NAME`                string, " +
                "`EMERGENCY_DIAGNOSE_CODE`            string, " +
                "`EMERGENCY_DIAGNOSE_NAME`            string, " +
                "`EMERGENCY_DATE`                     string, " +
                "`ADMIT_DIAGNOSE_DOCTOR`              string, " +
                "`ADMIT_DIAGNOSE_DOCTOR_CODE`         string, " +
                "`ADMISSION_DIAGNOSE_DOCTOR`          string, " +
                "`ADMISSION_DIAGNOSE_DOCTOR_CODE`     string, " +
                "`LEAVE_DIAGNOSE_DOCTOR`              string, " +
                "`LEAVE_DIAGNOSE_DOCTOR_CODE`         string, " +
                "`EMERGENCY_DIAGNOSE_DOCTOR`          string, " +
                "`EMERGENCY_DIAGNOSE_DOCTOR_CODE`     string, " +
                "`DEATH_DIAGNOSE_CODE`                string, " +
                "`DEATH_DIAGNOSE_NAME`                string, " +
                "`CREATE_TIME`                        timestamp, " +
                "`PUSH_STATUS`                        char" +
                ")" +
                "with (" +
                "   'connector'='jdbc'," +
                "   'url'='jdbc:mysql://192.168.31.206:3306/test_yuan'," +
                "   'table-name'='visit'," +
                "   'username'='root'," +
                "   'password'='password'," +
                // 设置数据的缓存事件，小时：hour，天：day
                "   'lookup.cache.ttl'='30 s'," +
                // 设置数据的缓存条数
                "   'lookup.cache.max-rows'='100'" +
                ")");
    }
}
